package com.shanks.framework.mq.rocketmq.consumer;

import com.shanks.framework.mq.common.constant.MQConstant;
import com.shanks.framework.mq.common.util.MQUtil;
import com.shanks.framework.mq.domain.Subscription;
import com.shanks.framework.mq.exception.MQException;
import com.shanks.framework.mq.processor.MQProcessor;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * rocketmq 消费者属性
 */
public class RocketMQConsumerAttribute {
    /**
     * 基础泳道namesrv
     */
    private String namesrvAddr;

    /**
     * 灰度泳道namesrv
     */
    private String canaryNamesrvAddr;

    /**
     * 实例名
     */
    private String instanceName;

    /**
     * 消费者组
     */
    private String groupName;

    /**
     * 消息消费最小线程数
     */
    private int consumeThreadMin;

    /**
     * 消息消费最大线程数
     */
    private int consumeThreadMax;

    /**
     * 消息模式（广播、集群）
     */
    private String messageModel;

    /**
     * 消息ack主题
     */
    private String ackMsgTopic = MQConstant.TOPIC.ACK_MSG_TOPIC;

    /**
     * 订阅关系 -- 消息处理器 MAP
     */
    private Map<Subscription, MQProcessor> sub2Process = new ConcurrentHashMap<>();

    /**
     * ack tags
     */
    private List<String> ackMsgTags = new CopyOnWriteArrayList<>();

    public String getNamesrvAddr() {
        return namesrvAddr;
    }

    public void setNamesrvAddr(String namesrvAddr) {
        this.namesrvAddr = namesrvAddr;
    }

    public String getCanaryNamesrvAddr() {
        return canaryNamesrvAddr;
    }

    public void setCanaryNamesrvAddr(String canaryNamesrvAddr) {
        this.canaryNamesrvAddr = canaryNamesrvAddr;
    }

    public String getInstanceName() {
        return instanceName;
    }

    public void setInstanceName(String instanceName) {
        this.instanceName = instanceName;
    }

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public int getConsumeThreadMin() {
        return consumeThreadMin;
    }

    public void setConsumeThreadMin(int consumeThreadMin) {
        this.consumeThreadMin = consumeThreadMin;
    }

    public int getConsumeThreadMax() {
        return consumeThreadMax;
    }

    public void setConsumeThreadMax(int consumeThreadMax) {
        this.consumeThreadMax = consumeThreadMax;
    }

    public String getMessageModel() {
        return messageModel;
    }

    public void setMessageModel(String messageModel) {
        this.messageModel = messageModel;
    }

    public String getAckMsgTopic() {
        return ackMsgTopic;
    }

    public void setAckMsgTopic(String ackMsgTopic) {
        this.ackMsgTopic = ackMsgTopic;
    }

    public Map<Subscription, MQProcessor> getSub2Process() {
        return sub2Process;
    }

    public List<String> getAckMsgTags() {
        return ackMsgTags;
    }

    /**
     * 订阅消息
     *
     * @param topic
     * @param tags
     * @param mqProcessor
     * @throws MQException
     */
    public void subscribe(String topic, List<String> tags, MQProcessor mqProcessor) throws MQException {
        String expression = StringUtils.join(tags, " || ");
        Subscription subscription = new Subscription(topic, expression);
        if (sub2Process.get(subscription) != null)
            throw new MQException("topic already exist: " + topic);
        sub2Process.putIfAbsent(subscription, mqProcessor);
    }


    /**
     * 订阅消息
     *
     * @param topic
     * @param tag
     * @param mqProcessor
     * @throws MQException
     */
    public void subscribe(String topic, String tag, MQProcessor mqProcessor) throws MQException {
        Subscription subscription = new Subscription(topic, tag);
        if (sub2Process.get(subscription) != null)
            throw new MQException("topic already exist: " + topic);
        sub2Process.putIfAbsent(subscription, mqProcessor);
    }

    /**
     * 订阅ack消息
     *
     * @param originTopic
     * @param originTags
     */
    public void subscribeAckMsg(String originTopic, List<String> originTags) {
        for (String tag : originTags)
            ackMsgTags.add(MQUtil.getAckMsgTag(originTopic, tag));
    }

    /**
     * 订阅ack消息
     *
     * @param originTopic
     * @param originTag
     */
    public void subscribeAckMsg(String originTopic, String originTag) {
        ackMsgTags.add(MQUtil.getAckMsgTag(originTopic, originTag));
    }
}
